Spark Streaming - Structured Streaming: Sliding Window Workbook

This workbook walks you through tracing sliding window streams with watermarking.

1. Sliding Window Parameters

• Window Size: 10 minutes
• Slide Duration: 5 minutes
• Watermark Delay: 10 minutes

2. Tasks

Task 1: Trace late arrivals

Trace how the streaming engine evaluates state across 3 incoming batches. Map records to windows and track the active watermark threshold.

Task 2: Implement Kafka to Delta Pipeline

Write the PySpark Structured Streaming code to read transactions from Kafka topic payments, parse the JSON payload explicitly, apply a 10-minute watermark on payment_time, and write the output in append mode to a Delta table at /mnt/delta/payments.

3. Step-by-Step Solutions

Solution 1: Stream Tracing Matrix

• Batch 1 (Incoming):
  • Record A: 12:05:00
  • Windows Triggered: [12:00-12:10], [12:05-12:15]
  • Max Event Time: 12:05:00
  • Active Watermark: None (initialized)
• Batch 2 (Incoming):
  • Record B: 12:20:00
  • Windows Triggered: [12:15-12:25], [12:20-12:30]
  • Max Event Time: 12:20:00
  • Active Watermark: 12:20:00 - 10 mins = 12:10:00
• Batch 3 (Incoming - Late):
  • Record C: 12:08:00
  • Evaluation: Is 12:08:00 > Active Watermark (12:10:00)?
  • Decision: No. The record is dropped as late data.

Solution 2: Kafka to Delta Code

from pyspark.sql.functions import from_json, col
from pyspark.sql.types import StructType, StructField, StringType, DoubleType, TimestampType

# Define payment payload schema
payment_schema = StructType([
    StructField("tx_id", StringType(), False),
    StructField("amount", DoubleType(), False),
    StructField("payment_time", TimestampType(), False)
])

# Read from Kafka stream
kafka_stream = spark.readStream \
    .format("kafka") \
    .option("kafka.bootstrap.servers", "localhost:9092") \
    .option("subscribe", "payments") \
    .load()

# Parse JSON and apply watermark
parsed_stream = kafka_stream.selectExpr("CAST(value AS STRING) as json_payload") \
    .select(from_json(col("json_payload"), payment_schema).alias("data")) \
    .select("data.*") \
    .withWatermark("payment_time", "10 minutes")

# Write to transactional Delta Table
query = parsed_stream.writeStream \
    .format("delta") \
    .outputMode("append") \
    .option("checkpointLocation", "/mnt/delta/payments/_checkpoints") \
    .start("/mnt/delta/payments")